package com.cnjson.sensor.nio;

import com.cnjson.sensor.protocolARP.Protocol;
import com.cnjson.sensor.protocolARP.waterQualitySensorProtocol.WaterQualitySensorProtocol;
import com.cnjson.sensor.protocolARP.waterQualitySensorProtocol.response.WaterQualitySensorResponse;
import com.cnjson.sensor.util.DataFormater;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

//import com.cnjson.common.utils.IdGen;

@Component
public class SocketServer {

    private static final Logger logger = LoggerFactory.getLogger(SocketServer.class);


    private Map<Integer, Integer> rejectedThreadCountMap = new ConcurrentHashMap<>();

    private RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
                int hashcode = r.hashCode();
                Integer count = rejectedThreadCountMap.get(hashcode);
                if (count == null) {
                    count = 0;
                    rejectedThreadCountMap.put(hashcode, count);
                } else {
                    count++;
                    rejectedThreadCountMap.put(hashcode, count);
                }
                if (count < 1) {
                    executor.execute(r);
                } else {
                    if (r instanceof WriteClientSocketHandler) {
                        WriteClientSocketHandler realThread = (WriteClientSocketHandler) r;
                        logger.info("服务系统繁忙,客户端WriteClientSocketHandler[" + realThread.client + "]请求被拒绝处理！");

                        SelectionKey selectionKey = realThread.client.keyFor(selector);
                        if (selectionKey != null) {
                            selectionKey.cancel();
                        }
                        if (realThread.client != null) {
                            try {
                                realThread.client.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        AtomicBoolean isWriting = isWritingMap.get(realThread.client);
                        isWriting.set(false);
                    } else if (r instanceof ReadClientSocketHandler) {
                        ReadClientSocketHandler realThread = (ReadClientSocketHandler) r;
                        logger.info("服务系统繁忙,客户端ReadClientSocketHandler[" + realThread.client + "]请求被拒绝处理！");
                        SelectionKey selectionKey = realThread.client.keyFor(selector);
                        if (selectionKey != null) {
                            selectionKey.cancel();
                        }
                        if (realThread.client != null) {
                            try {
                                realThread.client.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        AtomicBoolean isReading = isReadingMap.get(realThread.client);
                        isReading.set(false);
                    } else {
                        logger.info("服务系统繁忙,系统线程[" + r.getClass().getName() + "]被拒绝处理！");
                    }
                }
            } catch (InterruptedException e) {
                logger.info("RejectedExecutionHandler处理发生异常：" + e.getMessage());
            }
        }
    };

    private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(30, 50, 300, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(5), rejectedExecutionHandler);


    private static Map<String, SocketChannel> serverSocketMap = new ConcurrentHashMap<String, SocketChannel>();// 记录socket的键值对
    private static Map<SocketChannel, AtomicBoolean> isReadingMap = new ConcurrentHashMap<SocketChannel, AtomicBoolean>();
    private static Map<SocketChannel, AtomicBoolean> isWritingMap = new ConcurrentHashMap<SocketChannel, AtomicBoolean>();

    private static Selector selector = null;
    private ServerSocketChannel ss = null;
    private volatile boolean isClose = false;
    private static int status = 0;


    //根据不同协议进行解析
    private Map<String, Protocol> protocolMap = new HashMap<>();
    private WaterQualitySensorProtocol waterQualitySensorProtocol;

    @Value("${nio.port}")
    Integer port=2020;
    /**
     * 创建非阻塞服务器绑定2020端口
     */
    public SocketServer(WaterQualitySensorProtocol waterQualitySensorProtocol) {
        try {
            ss = ServerSocketChannel.open();
            ss.bind(new InetSocketAddress(port));
            ss.configureBlocking(false);
            selector = Selector.open();
            ss.register(selector, SelectionKey.OP_ACCEPT);


            //构造器自动注入
            this.waterQualitySensorProtocol=waterQualitySensorProtocol;
            protocolMap.put("waterQualitySensorProtocol", waterQualitySensorProtocol);
        } catch (Exception e) {
            logger.error(e.getMessage());
            close();
        }
    }


    public boolean isClose() {
        return isClose;
    }

    /**
     * 关闭服务器
     */
    private void close() {
        isClose = true;
        threadPool.shutdown();
        try {
            if (ss != null) {
                ss.close();
            }
            if (selector != null) {
                selector.close();
            }
        } catch (IOException e) {
            logger.error("服务器关闭发生异常：" + e.getMessage());
        }
    }

    /**
     * 启动选择器监听客户端事件
     */
    public void start() {
        threadPool.execute(new SelectorGuardHandler());
    }

    /**
     * Nio 通道选择器
     *
     * @author cgli
     */
    private class SelectorGuardHandler implements Runnable {

        @Override
        public void run() {

            while (!isClose) {
                logger.debug("正在轮询不要急。。。");
                try {
                    if (selector.select(10) == 0) {
                        continue;
                    }
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectedKey = iterator.next();
                        iterator.remove();
                        try {
                            if (selectedKey.isReadable()) {
                                SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
                                AtomicBoolean isReading = isReadingMap.get(socketChannel);
                                if (isReading == null) {
                                    isReading = new AtomicBoolean(false);
                                    isReadingMap.put(socketChannel, isReading);
                                }
                                while (isReading.get()) {
                                    Thread.sleep(5);
                                }
                                isReading.set(true);
                                threadPool.execute(new ReadClientSocketHandler(socketChannel));

                            } else if (selectedKey.isWritable()) {
                                Object responseMessage = selectedKey.attachment();
                                SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
                                selectedKey.interestOps(SelectionKey.OP_READ);
                                System.out.println("---------------------->isWritable");
                                logger.info("---------------------->isWritable");
                                threadPool.execute(new WriteClientSocketHandler(socketChannel, responseMessage));

                            } else if (selectedKey.isAcceptable()) {
                                ServerSocketChannel ssc = (ServerSocketChannel) selectedKey.channel();
                                SocketChannel clientSocket = ssc.accept();
                                clientSocket.configureBlocking(false);
                                logger.info("---------------------->isAcceptable");
                                clientSocket.register(selector, SelectionKey.OP_READ);
                            }
                        } catch (CancelledKeyException e) {
                            selectedKey.cancel();
                            logger.error("服务器启动或运行发生异常：" + e);
                        }
                    }
                } catch (Exception e) {
                    if (e instanceof NullPointerException) {
                        e.printStackTrace();
                        logger.error(e.getMessage());
                        close();
                    } else {
                        logger.error("服务器启动或运行发生异常：" + e);
                        close();
                    }
                    break;
                }
            }
        }
    }

    /**
     * 响应数据给客户端线程
     */
    private class WriteClientSocketHandler implements Runnable {
        private SocketChannel client;

        private Object responseMessage;

        private WriteClientSocketHandler(SocketChannel client, Object responseMessage) {
            this.client = client;
            this.responseMessage = responseMessage;
        }

        @Override
        public void run() {
            try {
                logger.info("------------------w1 准备写入数据-------------");
                byte[] responseByteData = null;
                String logResponseString = "";
                if (responseMessage instanceof byte[]) {
                    responseByteData = (byte[]) responseMessage;
                    logger.info("---------------w2 写字节数据-------------");
                    logResponseString = DataFormater.bytesToHexString(responseByteData);
                } else if (responseMessage instanceof String) {
                    logResponseString = (String) responseMessage;
                    responseByteData = logResponseString.getBytes();
                } else if (responseMessage != null) {
                    logger.info("不支持的数据类型" + responseMessage.getClass());
                    return;
                }
                if (responseByteData == null || responseByteData.length == 0) {
                    logger.info("服务器响应的写入数据为空");
                    return;
                }
                client.write(ByteBuffer.wrap(responseByteData));

                logger.info("服务端响应客户端[" + client.getRemoteAddress() + "]写入数据 :[" + logResponseString + "]");
            } catch (Exception e) {
                try {
                    logger.info("server响应客户端[" + client.getRemoteAddress() + "]数据 异常[" + e.getMessage() + "]");
                    SelectionKey selectionKey = client.keyFor(selector);
                    if (selectionKey != null) {
                        selectionKey.cancel();
                    }
                    if (client != null) {
                        client.close();
                    }
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            } finally {
                AtomicBoolean isWriting = isWritingMap.get(client);
                isWriting.set(false);
            }
        }
    }

    /**
     * 读客户端发送数据线程
     */


    private class ReadClientSocketHandler implements Runnable {

        private SocketChannel client;

        private ReadClientSocketHandler(SocketChannel client) {
            this.client = client;
        }

        @Override
        public void run() {
            //这里后面如果有多个协议，可以在这里先读出协议的标识位（标识哪个协议的），然后在执行对应的协议解析工作

            protocolMap.get("waterQualitySensorProtocol").read(client, selector, serverSocketMap,
                    isWritingMap, isReadingMap);

        }


    }



    /**
     * 后面这个方法是不放在这个类的
     * @param electricSignal 电信号（0x02高电平,0x82低电平，0x81 正向脉冲 开,0x01 反向脉冲 停），等等
     * @param remote         设备的网络ip和端口  例如：  /39.144.18.93:56936
     * @param equipmentNum   设备id
     * @return
     */
    public boolean controllEquipment(String electricSignal, String remote, int equipmentNum) {
        try {

            String equipmentHex = Integer.toHexString(equipmentNum);
            String high = equipmentHex.substring(0, 2);
            String low = equipmentHex.substring(2, 4);
            byte hb = (byte) Integer.parseInt(high, 16);
            byte lb = (byte) Integer.parseInt(low, 16);
            byte[] content = {0};
            byte[] device = {hb, lb, 0, 0};
            byte[] request = WaterQualitySensorResponse.buildRequestPackage((byte) Integer.parseInt(electricSignal, 16), device, content);

            SocketChannel client = null;
            if (SocketServer.serverSocketMap != null) {
                client = SocketServer.serverSocketMap.get(remote);
            }


            if (client != null) {
                AtomicBoolean isWriting = SocketServer.isWritingMap.get(client);
                if (isWriting == null) {
                    isWriting = new AtomicBoolean(false);
                    SocketServer.isWritingMap.put(client, isWriting);
                }
                while (isWriting != null && isWriting.get()) {
                    Thread.sleep(5L);
                }
                if (isWriting != null) {
                    isWriting.set(true);
                }
                client.register(SocketServer.selector, 4, request);
            }


        } catch (Exception e) {
            SocketServer.logger.error("________________异常————————————————————————" + e);
        }
        return false;
    }




}
